package com.sunday.common.mq.rocket.brave.study.apache.e7_Brave_Trace_Rocket_Pull;

import com.sunday.common.mq.rocket.brave.annotation.TracingRocketHandler;
import com.sunday.common.mq.rocket.brave.study.apache.User;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;
import java.util.stream.Stream;

/**
 * 邮件跟踪
 * https://github.com/apache/rocketmq-spring/wiki/Message-Tracing
 */
@Slf4j
@RestController
@RequestMapping("/test")
@RequiredArgsConstructor
public class BraveTraceApi {

    final RocketMQTemplate rocketMQTemplate;

    @GetMapping("send")
    public void send() {
        Stream.iterate(0, i -> i < 1, i -> i + 1).forEach(i -> {
            User user = User.builder().id(i).userName("john" + i).userAge(33 + i).build();
            log.info("{}", user);
            rocketMQTemplate.convertAndSend("your_topic", user);
        });
    }

    @TracingRocketHandler
    @GetMapping("pull")
    public void pull() throws InterruptedException {
        log.info("start----------------------------");
        List<MessageExt> list;
        /**
         * 特殊说一下 一般来说 rocketMQTemplate.receive 获取到消息非常的迅速，
         * 但是次样例中 存在一个滞后的情况，引发的原因是 receive(Class<T> clazz, long timeout) 里面 timeout 带有5秒的空闲
         * 如果想避免请 设置 timeout
         */
        while (!(list = rocketMQTemplate.receive(MessageExt.class, 0)).isEmpty()) {
//        List<User> list;
//        while (!(list = rocketMQTemplate.receive(User.class)).isEmpty()) {
            log.info("{}", list.size());
            list.stream().forEach(value -> {
//                log.info("Receive New Messages : {}", value);
                log.info("Receive body : {}", new String(value.getBody()));
            });
//            TimeUnit.SECONDS.sleep(5);
        }
        log.info("end----------------------------");
    }

}